FastAPI এর Background Tasks এবং Async Programming

FastAPI একটি আধুনিক Python framework যা Async Programming এবং Background Tasks সমর্থন করে। এটি দ্রুত এবং স্কেলেবল অ্যাপ্লিকেশন তৈরি করতে সহায়ক। এখানে আমরা জানব কিভাবে Background Tasks এবং Async Programming FastAPI তে ব্যবহৃত হয় এবং এগুলোর মাধ্যমে কিভাবে অ্যাসিঙ্ক্রোনাস কোড পরিচালনা করা যায়।

Async Programming কি?

Async Programming হল এমন একটি প্রোগ্রামিং প্যাটার্ন যা আপনাকে ব্লকিং অপারেশনগুলোকে সিঙ্ক্রোনাসভাবে না করে অ্যাসিঙ্ক্রোনাসভাবে চালাতে সাহায্য করে। এর মাধ্যমে একাধিক টাস্ক একসাথে পরিচালনা করা যায়, যা অ্যাপ্লিকেশনকে আরও দ্রুত এবং স্কেলেবল করে তোলে। FastAPI অ্যাসিঙ্ক্রোনাস কোড পরিচালনা করতে async/await কিওয়ার্ড ব্যবহার করে।

Async Programming FastAPI তে

FastAPI স্বাভাবিকভাবে asyncio লাইব্রেরির উপর ভিত্তি করে কাজ করে, যা অ্যাসিঙ্ক্রোনাস কোডের জন্য একটি পপুলার লাইব্রেরি। FastAPI অ্যাসিঙ্ক্রোনাস ফাংশনগুলো async def দিয়ে ডিফাইন করে, যা await কিওয়ার্ড ব্যবহার করে ব্যাকগ্রাউন্ড অপারেশনগুলোকে অ্যাসিঙ্ক্রোনাসভাবে চালনা করতে সহায়ক হয়।

উদাহরণ: Async Programming

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Async Function
async def fake_database_query():
    await asyncio.sleep(3)
    return {"message": "Data fetched from database"}

@app.get("/data/")
async def get_data():
    data = await fake_database_query()  # Async call
    return data

এখানে, fake_database_query একটি asynchronous ফাংশন যা 3 সেকেন্ড সময় নেয়, এবং await কিওয়ার্ড ব্যবহার করে এটি অপেক্ষা করা হয়। get_data ফাংশনটিও asynchronous, এবং এতে await fake_database_query() ব্যবহার করে অ্যাসিঙ্ক্রোনাস ফাংশনটি কল করা হয়।

রিকোয়েস্ট উদাহরণ:

GET /data/

রেসপন্স:

{
  "message": "Data fetched from database"
}

Background Tasks

Background Tasks হল এমন কাজ যা আপনি অ্যাপ্লিকেশন চলাকালীন সময়ে ব্যাকগ্রাউন্ডে চালাতে পারেন, যেমন ইমেইল পাঠানো, লগ লেখার কাজ, বা ফাইল প্রক্রিয়া করা। FastAPI তে Background Tasks ব্যবস্থাপনা করতে BackgroundTasks ক্লাস ব্যবহার করা হয়।

উদাহরণ: Background Tasks

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

# Background Task Function
def write_log(message: str):
    time.sleep(5)  # Simulate a time-consuming task
    with open("log.txt", mode="a") as log:
        log.write(f"{message}\n")

@app.get("/send-email/")
async def send_email(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Email sent to user")  # Add task to background
    return {"message": "Email is being sent in the background"}

এখানে, write_log একটি ব্যাকগ্রাউন্ড টাস্ক যা 5 সেকেন্ড সময় নেয়। send_email এন্ডপয়েন্টে যখন কল করা হয়, তখন এই ব্যাকগ্রাউন্ড টাস্কটি background_tasks.add_task() মাধ্যমে ব্যাকগ্রাউন্ডে চলে যায় এবং ফাংশনটি অবিলম্বে রেসপন্স দেয়।

রিকোয়েস্ট উদাহরণ:

GET /send-email/

রেসপন্স:

{
  "message": "Email is being sent in the background"
}

ব্যাকগ্রাউন্ডে log.txt ফাইলে লেখার উদাহরণ:

Email sent to user

ব্যাকগ্রাউন্ড টাস্ক ফাংশন write_log 5 সেকেন্ড পরে log.txt ফাইলে ডাটা লেখে, কিন্তু ক্লায়েন্টকে অবিলম্বে রেসপন্স পাঠানো হয়।

Background Tasks এবং Async Programming একসাথে ব্যবহৃত হওয়া

FastAPI তে Background Tasks এবং Async Programming একসাথে ব্যবহার করা সম্ভব। আপনি async ফাংশনগুলিকে ব্যাকগ্রাউন্ড টাস্ক হিসেবে যোগ করতে পারেন এবং এই দুইটি কার্যকারিতা একসাথে কার্যকরীভাবে পরিচালনা করতে পারেন।

উদাহরণ: Background Tasks এবং Async Programming

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

# Async function that simulates time-consuming task
async def fake_long_task():
    await asyncio.sleep(5)
    return "Task completed"

# Background task that triggers an async function
def background_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(fake_long_task)

@app.get("/run-task/")
async def run_task(background_tasks: BackgroundTasks):
    background_task(background_tasks)  # Add async task to background
    return {"message": "Task is running in the background"}

এখানে, fake_long_task একটি async function, এবং background_task ফাংশনটি এটি ব্যাকগ্রাউন্ডে চালাতে background_tasks.add_task() ব্যবহার করে। যখন /run-task/ এন্ডপয়েন্টে রিকোয়েস্ট করা হয়, তখন ব্যাকগ্রাউন্ডে দীর্ঘকালীন কাজটি চলে যায় এবং ক্লায়েন্টকে অবিলম্বে রেসপন্স দেওয়া হয়।

রিকোয়েস্ট উদাহরণ:

GET /run-task/

রেসপন্স:

{
  "message": "Task is running in the background"
}

ব্যাকগ্রাউন্ডে 5 সেকেন্ড পর fake_long_task ফাংশন সম্পন্ন হবে।

FastAPI এর Async/await এবং Background Tasks এর সুবিধা

1. স্কেলেবিলিটি:
   FastAPI অ্যাসিঙ্ক্রোনাস কোডের মাধ্যমে একাধিক কাজ একসাথে চালাতে পারে, যার ফলে অ্যাপ্লিকেশনটি অনেক বেশি স্কেলেবল হয়।
2. পূর্ণ রেসপন্স:
   ব্যাকগ্রাউন্ড টাস্কগুলো অ্যাসিঙ্ক্রোনাসভাবে কাজ করলেও, ক্লায়েন্টের কাছে দ্রুত রেসপন্স প্রদান করা সম্ভব।
3. একমাত্রিক টাস্ক:
   ব্যাকগ্রাউন্ড টাস্কের মাধ্যমে আপনি টাইম-তীব্র কাজগুলো ব্যাকগ্রাউন্ডে চালাতে পারেন, যেমন ইমেইল পাঠানো, লগ লেখা, বা ফাইল আপলোড প্রক্রিয়া করা, যা ক্লায়েন্টের অভিজ্ঞতা উন্নত করে।

FastAPI তে Async Programming এবং Background Tasks ব্যবস্থাপনা খুবই সহজ এবং কার্যকর। Async ফাংশনগুলোর মাধ্যমে আপনি অ্যাসিঙ্ক্রোনাস কার্যক্রম পরিচালনা করতে পারেন, এবং Background Tasks ব্যবহার করে টাইম-কনজ্যুমিং কাজগুলো ব্যাকগ্রাউন্ডে চালাতে পারেন। এই সুবিধাগুলো FastAPI কে দ্রুত, স্কেলেবল এবং কার্যকরী অ্যাপ্লিকেশন তৈরি করতে সহায়ক করে তোলে।

Asynchronous Programming (অ্যাসিঙ্ক্রোনাস প্রোগ্রামিং) হল একটি প্রোগ্রামিং প্যারাডাইম যেখানে টাইম-থ্রেটিং এবং দীর্ঘ সময় ধরে চলা কাজগুলি (যেমন I/O অপারেশন, ডাটাবেস কুয়েরি, API কল ইত্যাদি) অন্য কোনো কাজ বাধা না দিয়ে সম্পন্ন করা হয়। Python তে asyncio লাইব্রেরি এবং async/await কিওয়ার্ড ব্যবহার করে অ্যাসিঙ্ক্রোনাস প্রোগ্রামিং করা যায়। FastAPI এই ধরনের প্রোগ্রামিং সহজেই সমর্থন করে, যা দ্রুত এবং স্কেলেবল API তৈরি করতে সাহায্য করে।

FastAPI এবং Asynchronous Programming

FastAPI হল একটি ASGI (Asynchronous Server Gateway Interface) ফ্রেমওয়ার্ক, যা synchronous এবং asynchronous কোড উভয়ই সমর্থন করে। FastAPI অ্যাসিঙ্ক্রোনাস প্রোগ্রামিংয়ের সুবিধা নিতে async এবং await কিওয়ার্ড ব্যবহার করে কার্যকলাপ সম্পন্ন করতে পারে, যেমন ডাটাবেস থেকে ডাটা পড়া, থার্ড-পার্টি API কল করা, ফাইল I/O ইত্যাদি।

Asynchronous Programming এর সুবিধা

1. দ্রুত পারফরম্যান্স: অ্যাসিঙ্ক্রোনাস কোড অনেক দ্রুত এবং কম সিস্টেম রিসোর্স ব্যবহার করে।
2. স্কেলেবিলিটি: অ্যাসিঙ্ক্রোনাস প্রোগ্রামিং একাধিক I/O অপারেশনকে একযোগে প্রক্রিয়া করতে পারে, যার ফলে ওয়েব অ্যাপ্লিকেশন উচ্চ লোড সহ ভালভাবে স্কেল করতে সক্ষম হয়।
3. এফিশিয়েন্ট রিসোর্স ব্যবহারের ক্ষমতা: সিঙ্ক্রোনাস কোডের তুলনায় অ্যাসিঙ্ক্রোনাস কোড অধিক কার্যক্ষমভাবে সিস্টেম রিসোর্স ব্যবহার করে।

FastAPI তে Asynchronous Programming এর ব্যবহার

FastAPI তে asynchronous কোড লেখার জন্য async এবং await কিওয়ার্ড ব্যবহার করা হয়। এটি মূলত I/O-bound অপারেশনের জন্য উপযুক্ত, যেমন ডাটাবেস কোয়েরি, API কল, এবং ফাইল হ্যান্ডলিং।

উদাহরণ ১: Asynchronous Route

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Asynchronous Route
@app.get("/")
async def read_root():
    await asyncio.sleep(1)  # Simulating a non-blocking task
    return {"message": "Hello, FastAPI!"}

এখানে, async def দিয়ে রাউটটি ডিফাইন করা হয়েছে এবং await asyncio.sleep(1) ব্যবহৃত হয়েছে যা এক সেকেন্ড অপেক্ষা করবে কিন্তু অন্য কোনো কাজের জন্য ব্লক করবে না। এটি অ্যাসিঙ্ক্রোনাস কোডের একটি সাধারণ উদাহরণ।

রিকোয়েস্ট উদাহরণ:

GET /
রেসপন্স:

{
  "message": "Hello, FastAPI!"
}

এখানে, await asyncio.sleep(1) ব্যবহারের ফলে সার্ভার এক সেকেন্ড অপেক্ষা করে, কিন্তু এটি অন্য রিকোয়েস্ট ব্লক করবে না। আপনি অন্যান্য রিকোয়েস্ট গ্রহণ করতে পারবেন।

Asynchronous I/O Operation (ডাটাবেস বা API কল)

FastAPI-তে অ্যাসিঙ্ক্রোনাস ফাংশন ব্যবহার করে আপনি ডাটাবেস থেকে ডাটা পড়তে বা API কল করতে পারেন, যেটি I/O-bound অপারেশন। এখানে, আমরা দেখব কিভাবে অ্যাসিঙ্ক্রোনাস API কল করা যায়।

উদাহরণ ২: Asynchronous Database Query (Simulated)

from fastapi import FastAPI
import asyncio

app = FastAPI()

# Simulating an asynchronous database query
async def get_item_from_db(item_id: int):
    await asyncio.sleep(2)  # Simulate delay
    return {"item_id": item_id, "name": "Laptop", "price": 1500.00}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = await get_item_from_db(item_id)
    return item

এখানে, get_item_from_db একটি অ্যাসিঙ্ক্রোনাস ফাংশন, যা ডাটাবেস থেকে ডাটা পড়তে ২ সেকেন্ড সময় নিবে, কিন্তু অন্যান্য রিকোয়েস্ট প্রসেস করা যাবে। await ব্যবহার করে এই ফাংশনটি ডাটাবেস কোয়েরি শেষ না হওয়া পর্যন্ত অন্য কাজ করতে বাধা সৃষ্টি করবে না।

রিকোয়েস্ট উদাহরণ:

GET /items/123
রেসপন্স:

{
  "item_id": 123,
  "name": "Laptop",
  "price": 1500.00
}

এখানে, await get_item_from_db(item_id) ব্যবহার করা হয়েছে যাতে ডাটাবেস কোয়েরি চলাকালীন সার্ভার অন্য কাজ করতে পারে।

Asynchronous File I/O Operation

FastAPI-তে অ্যাসিঙ্ক্রোনাস ফাইল I/O অপারেশনও করা সম্ভব, যেখানে আপনি ফাইল পড়া বা লেখার সময় সার্ভারের রেসপন্স ব্লক না করে অন্যান্য কাজ চালাতে পারেন।

উদাহরণ ৩: Asynchronous File Reading

from fastapi import FastAPI
import aiofiles  # For asynchronous file handling

app = FastAPI()

@app.get("/readfile/")
async def read_file():
    async with aiofiles.open('sample.txt', mode='r') as file:
        content = await file.read()
    return {"file_content": content}

এখানে aiofiles ব্যবহার করা হয়েছে, যা ফাইল অপারেশনকে অ্যাসিঙ্ক্রোনাসভাবে পরিচালনা করে, যাতে সার্ভার কোনো ফাইলের সাথে কাজ করার সময় ব্লক না হয়ে অন্য রিকোয়েস্ট গ্রহণ করতে পারে।

রিকোয়েস্ট উদাহরণ:

GET /readfile/
রেসপন্স:

{
  "file_content": "This is a sample file content."
}

Asynchronous API Calls

FastAPI তে async ও await ব্যবহার করে আপনি একাধিক API কলও অ্যাসিঙ্ক্রোনাসভাবে করতে পারেন, যা I/O-bound অপারেশনকে দ্রুত সম্পন্ন করতে সাহায্য করে।

উদাহরণ ৪: Asynchronous HTTP Requests

import httpx
from fastapi import FastAPI

app = FastAPI()

# Asynchronous API call using httpx
@app.get("/external-api/")
async def get_external_data():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
    return response.json()

এখানে httpx.AsyncClient() ব্যবহার করে অ্যাসিঙ্ক্রোনাস HTTP রিকোয়েস্ট করা হয়েছে, যা await ব্যবহার করে রেসপন্স পাওয়ার আগে অন্য কাজ করার সুযোগ দেয়।

Asynchronous Programming-এর সুবিধা

1. দ্রুত এবং স্কেলেবল: অ্যাসিঙ্ক্রোনাস কোড উচ্চ লোড সহ ভালভাবে কাজ করতে সক্ষম, কারণ এটি একাধিক কাজ একযোগে করতে পারে।
2. কম রিসোর্স ব্যবহার: অ্যাসিঙ্ক্রোনাস প্রোগ্রামিং সিস্টেমের রিসোর্সের ব্যবহার অধিক কার্যকর করে, বিশেষ করে I/O-bound অপারেশনগুলো।
3. থ্রেড ব্যবহার কমানো: অ্যাসিঙ্ক্রোনাস প্রোগ্রামিং থ্রেড-ভিত্তিক সিস্টেমের চেয়ে কম রিসোর্স ব্যবহার করে, যার ফলে সিস্টেমের পারফরম্যান্স বাড়ে।

FastAPI তে Asynchronous Programming ব্যবহারের মাধ্যমে আপনি উচ্চ পারফরম্যান্স এবং স্কেলেবিলিটি নিশ্চিত করতে পারেন। async এবং await কিওয়ার্ড ব্যবহার করে, অ্যাসিঙ্ক্রোনাস HTTP রিকোয়েস্ট, ডাটাবেস কোয়েরি, ফাইল I/O এবং অন্যান্য I/O-bound কাজগুলি দ্রুত এবং কম রিসোর্স ব্যবহার করে সম্পন্ন করা যায়। FastAPI এই ধরনের প্রোগ্রামিং খুব সহজ করে তোলে, যা ডেভেলপারদের দ্রুত এবং কার্যকর API তৈরি করতে সাহায্য করে।

FastAPI হল একটি Asynchronous (আসিঙ্ক্রোনাস) ওয়েব ফ্রেমওয়ার্ক, যা async এবং await ব্যবহার করে কোডের কার্যক্ষমতা এবং স্কেলেবিলিটি বৃদ্ধি করতে সক্ষম। Async এবং Await ব্যবহার করলে আপনার অ্যাপ্লিকেশন একাধিক কাজ একসাথে (concurrently) সম্পাদন করতে পারে, যার ফলে পারফরম্যান্স অনেক উন্নত হয়, বিশেষ করে বড় লোডের অ্যাপ্লিকেশনগুলির জন্য।

এখানে আলোচনা করা হবে কিভাবে async এবং await FastAPI তে কাজ করে এবং এর উপকারিতা কী।

Async এবং Await কী?

• Async (Asynchronous):
  Async হলো এমন একটি প্রোগ্রামিং মডেল যা I/O অপারেশন (যেমন, নেটওয়ার্ক কল, ডাটাবেস অপারেশন) করার সময় কোডের execution থামিয়ে না রেখে, অন্য কাজ করতে দেয়।
• Await:
  Await হল একটি কিওয়ার্ড যা async ফাংশনের মধ্যে ব্যবহার করা হয়। এটি কোনো Asynchronous অপারেশন সম্পন্ন হওয়ার জন্য অপেক্ষা করতে বলে, কিন্তু অন্যান্য কোড চলতে থাকে।

উদাহরণ:

async def my_function():
    # কিছু আসিঙ্ক্রোনাস কাজ করা হচ্ছে
    result = await some_async_function()
    return result

FastAPI তে Async এবং Await ব্যবহার করা

FastAPI তে async এবং await এর মাধ্যমে কোড অনেক দ্রুত এবং স্কেলেবল হয়ে ওঠে, বিশেষ করে যখন API কে অনেক রিকোয়েস্ট পরিচালনা করতে হয়।

FastAPI তে Async Function ব্যবহার:

FastAPI তে async def দিয়ে asynchronous view function বা endpoint তৈরি করা হয়। যখন আপনি await ব্যবহার করেন, তখন FastAPI অন্য রিকোয়েস্ট প্রসেস করার সময় আপনার I/O অপারেশনটি সম্পন্ন হওয়ার জন্য অপেক্ষা করে।

উদাহরণ: Async Function FastAPI তে

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def fetch_data():
    await asyncio.sleep(2)  # এখানে ২ সেকেন্ড অপেক্ষা করবে
    return {"data": "Hello, World!"}

@app.get("/async-example")
async def async_example():
    result = await fetch_data()  # await করে ফাংশনটির কাজ শেষ হওয়ার জন্য
    return result

এখানে, fetch_data() একটি async function যা ২ সেকেন্ডের জন্য অপেক্ষা করে (উদাহরণস্বরূপ, নেটওয়ার্ক কল বা ডাটাবেস রিডিং)। await fetch_data() এর মাধ্যমে FastAPI এটিকে কল করে এবং অন্যান্য রিকোয়েস্ট প্রসেস করার সময় এটি অপেক্ষা করে।

রিকোয়েস্ট উদাহরণ:

GET /async-example

রেসপন্স (কিছু সময় পর):

{
  "data": "Hello, World!"
}

Async এবং Await এর উপকারিতা

1. উচ্চ পারফরম্যান্স এবং স্কেলেবিলিটি:
   Async এবং Await ব্যবহার করার মাধ্যমে কোড একাধিক কাজ একসাথে করতে পারে, যেমন, ডাটাবেসের সাথে একাধিক কনকারেন্ট রিকোয়েস্ট হ্যান্ডল করা। এর ফলে সার্ভারের প্রতি রিকোয়েস্টের প্রতিক্রিয়া সময় (response time) কমে এবং আরও বেশি ইউজার হ্যান্ডল করা সম্ভব হয়।
2. Non-blocking I/O অপারেশন:
   সাধারণ synchronous ফাংশনে যখন একটি I/O অপারেশন চলতে থাকে (যেমন, নেটওয়ার্ক রিকোয়েস্ট বা ডাটাবেস কল), পুরো কোড থেমে থাকে। কিন্তু Async কোডের মধ্যে, যখন একটি অপারেশন সম্পন্ন হওয়ার জন্য অপেক্ষা করা হয়, তখন অন্য কাজগুলি চলতে থাকে। এটি উচ্চ ট্র্যাফিক বা বড় অ্যাপ্লিকেশনে কার্যক্ষমতা বৃদ্ধি করে।
3. কম সময় অপেক্ষা করা:
   Async কোড অপেক্ষার সময় কমাতে সাহায্য করে। যেমন, যখন ডাটাবেসে বা API-তে একাধিক রিকোয়েস্ট করা হয়, তখন তারা একে অপরকে ব্লক না করে একসাথে কাজ করতে পারে।

Synchronous vs Asynchronous: কী পার্থক্য?

• Synchronous:
  একে একে কাজ করে, প্রতিটি ফাংশন পরবর্তী ফাংশন শুরু করার আগে শেষ হতে হবে।

  উদাহরণ:

  def get_data():
      result = db_query()  # থামিয়ে রাখে যখন ডাটাবেস থেকে ডাটা আসে না
      return result
  

• Asynchronous:
  একসাথে একাধিক কাজ সম্পন্ন করতে সক্ষম। যখন কোনো একক কাজ শেষ হতে সময় নেয়, তখন অন্য কাজ করতে পারে।

  উদাহরণ:

  async def get_data():
      result = await db_query()  # অন্য কাজ চলতে থাকে যখন ডাটাবেস থেকে ডাটা আসে
      return result
  

Real-world Example: Database Querying with Async

ধরা যাক, আমরা একটি ডাটাবেসে কুইরি পাঠাচ্ছি এবং আমরা চাই যে আমাদের API তে একাধিক রিকোয়েস্ট হ্যান্ডল করার সময় ডাটাবেস থেকে ডাটা সংগ্রহ করা হোক।

from fastapi import FastAPI
import databases  # Async DB client

DATABASE_URL = "sqlite:///./test.db"

database = databases.Database(DATABASE_URL)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/users/")
async def get_users():
    query = "SELECT * FROM users"
    results = await database.fetch_all(query)  # Async DB query
    return results

এখানে, আমরা databases প্যাকেজ ব্যবহার করে ডাটাবেস থেকে অ্যাসিঙ্ক্রোনাস কোয়েরি করছি। await ব্যবহার করে FastAPI অপেক্ষা করছে, তবে অন্যান্য রিকোয়েস্ট হ্যান্ডলিং চলতে থাকে।

Error Handling with Async

FastAPI তে asynchronous কোড ব্যবহারের সময়, আপনি সাধারণভাবে try-except ব্লক ব্যবহার করতে পারেন এবং অ্যাসিঙ্ক্রোনাস ফাংশনেও async ত্রুটি পরিচালনা করতে পারেন।

@app.get("/fetch-data/")
async def fetch_data():
    try:
        result = await fetch_data_from_db()
        return result
    except Exception as e:
        return {"error": str(e)}

FastAPI তে async এবং await ব্যবহারের মাধ্যমে অ্যাপ্লিকেশন আরও দ্রুত এবং স্কেলেবল হয়ে ওঠে। আপনি যখন I/O-বাঁধিত কাজ (যেমন ডাটাবেস অ্যাক্সেস, নেটওয়ার্ক কল) করবেন, তখন async এবং await ব্যবহারের মাধ্যমে আপনার কোড আরও কার্যকর হবে এবং সার্ভারের পারফরম্যান্স বৃদ্ধি পাবে। FastAPI এর সাথে asynchronous programming আপনাকে উচ্চ লোড সহ API তৈরি করতে সহায়ক, যা সিস্টেমের রেসপন্স টাইম এবং ক্ষমতা বাড়াতে সাহায্য করে।

FastAPI তে Background Tasks ব্যবস্থাপনা একটি শক্তিশালী বৈশিষ্ট্য, যা আপনাকে API রিকোয়েস্টের প্রক্রিয়াকরণের সময় ব্যাকগ্রাউন্ডে কিছু কাজ চালানোর সুযোগ দেয়। এটি বিশেষভাবে তখন উপকারী যখন আপনি দীর্ঘ সময় ধরে চলা কাজ (যেমন ইমেইল পাঠানো, ডাটা প্রসেসিং ইত্যাদি) করতে চান, তবে ক্লায়েন্টকে দ্রুত রেসপন্স দেওয়ার প্রয়োজন।

FastAPI তে BackgroundTasks ব্যবহার করে আপনি রিকোয়েস্টের সাথে সম্পর্কিত কাজগুলো ব্যাকগ্রাউন্ডে চালাতে পারেন, যাতে প্রধান অ্যাপ্লিকেশন থ্রেড ব্লক না হয়।

Step 1: BackgroundTasks Import করা

FastAPI-তে BackgroundTasks ব্যবহারের জন্য আপনাকে প্রথমে BackgroundTasks ক্লাসটি ইনপোর্ট করতে হবে।

from fastapi import FastAPI, BackgroundTasks

Step 2: Background Task ফাংশন তৈরি করা

আপনি যেসব কাজ ব্যাকগ্রাউন্ডে করতে চান, সেগুলোর জন্য একটি আলাদা ফাংশন তৈরি করুন। এটি সাধারণত একটি সিঙ্ক্রোনাস ফাংশন হবে যা ব্যাকগ্রাউন্ডে কার্যকরী হবে।

উদাহরণ: ব্যাকগ্রাউন্ড টাস্ক ফাংশন

import time

def send_email(email: str):
    time.sleep(5)  # সিমুলেট করা হচ্ছে একটি সময়সাপেক্ষ কাজ
    print(f"Email sent to {email}")

এখানে, send_email ফাংশনটি একটি ইমেইল পাঠানোর কাজ সিমুলেট করছে, যেটি ৫ সেকেন্ড সময় নেবে।

Step 3: BackgroundTasks ব্যবহার করা

এখন, FastAPI-র মধ্যে BackgroundTasks ব্যবহার করে এই ব্যাকগ্রাউন্ড ফাংশনটি কার্যকরী করতে হবে।

উদাহরণ: Background Task রিকোয়েস্টে যুক্ত করা

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str):
    time.sleep(5)
    print(f"Email sent to {email}")

@app.get("/send-notification/")
async def send_notification(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email)
    return {"message": "Notification sent in the background"}

এখানে, /send-notification/ রিকোয়েস্টে email প্যারামিটারটি ব্যাকগ্রাউন্ডে ইমেইল পাঠানোর কাজ শুরু করবে। background_tasks.add_task() ফাংশনটি ব্যাকগ্রাউন্ডে ফাংশনটি রান করানোর জন্য ব্যবহৃত হচ্ছে।

Step 4: Request এবং Background Task একসাথে ব্যবস্থাপনা

এখন যখন রিকোয়েস্ট আসবে, তখন ক্লায়েন্ট দ্রুত রেসপন্স পাবে, কিন্তু ইমেইল পাঠানোর কাজ ব্যাকগ্রাউন্ডে চলতে থাকবে।

রিকোয়েস্ট উদাহরণ:

GET /send-notification/?email=example@example.com

রেসপন্স:

{
  "message": "Notification sent in the background"
}

এখানে, Email sent to example@example.com বার্তা ব্যাকগ্রাউন্ডে কনসোলে প্রদর্শিত হবে, কিন্তু ক্লায়েন্টকে অপেক্ষা করতে হবে না।

Step 5: Multiple Background Tasks

FastAPI তে আপনি একাধিক ব্যাকগ্রাউন্ড টাস্কও যুক্ত করতে পারেন। নিচে একটি উদাহরণ দেওয়া হলো যেখানে একাধিক ব্যাকগ্রাউন্ড টাস্ক করা হচ্ছে।

উদাহরণ: একাধিক ব্যাকগ্রাউন্ড টাস্ক

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def send_email(email: str):
    time.sleep(5)
    print(f"Email sent to {email}")

def send_sms(phone: str):
    time.sleep(3)
    print(f"SMS sent to {phone}")

@app.get("/send-notifications/")
async def send_notifications(background_tasks: BackgroundTasks, email: str, phone: str):
    background_tasks.add_task(send_email, email)
    background_tasks.add_task(send_sms, phone)
    return {"message": "Notifications sent in the background"}

এখানে, send_email এবং send_sms দুটি ব্যাকগ্রাউন্ড টাস্ক একসাথে কার্যকরী করা হচ্ছে।

রিকোয়েস্ট উদাহরণ:

GET /send-notifications/?email=example@example.com&phone=1234567890

রেসপন্স:

{
  "message": "Notifications sent in the background"
}

এই রিকোয়েস্টের মাধ্যমে দুটি কাজ ব্যাকগ্রাউন্ডে চলে যাবে, কিন্তু ক্লায়েন্ট দ্রুত রেসপন্স পাবে।

Step 6: Background Task Logging

ব্যাকগ্রাউন্ড টাস্কের ফলাফল বা প্রগ্রেস ট্র্যাক করতে আপনি লগিং ব্যবহার করতে পারেন।

উদাহরণ: ব্যাকগ্রাউন্ড টাস্ক লগিং

import logging
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

logging.basicConfig(level=logging.INFO)

def send_email(email: str):
    logging.info(f"Sending email to {email}...")
    time.sleep(5)
    logging.info(f"Email sent to {email}")

@app.get("/send-notification/")
async def send_notification(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email)
    return {"message": "Notification sent in the background"}

এখানে, ব্যাকগ্রাউন্ড টাস্কের প্রক্রিয়ার প্রতিটি ধাপে লগিং করা হয়েছে।

FastAPI তে Background Tasks ব্যবস্থাপনা খুবই শক্তিশালী এবং সহজ। এটি আপনার API-কে দ্রুত রেসপন্স দিতে সহায়তা করে, বিশেষত যখন আপনাকে দীর্ঘ সময় ধরে চলা কাজ করতে হয়। BackgroundTasks ব্যবহার করে আপনি ব্যাকগ্রাউন্ডে বিভিন্ন কাজ চালাতে পারেন, যেমন ইমেইল পাঠানো, ডাটা প্রসেসিং, অথবা সেকেন্ডারি কাজগুলো। এতে আপনার API এর কার্যক্ষমতা বৃদ্ধি পায় এবং ক্লায়েন্ট দ্রুত রেসপন্স পায়।

FastAPI হল একটি অত্যন্ত দ্রুত এবং কার্যকরী ওয়েব ফ্রেমওয়ার্ক যা asynchronous প্রোগ্রামিং সমর্থন করে। এর সাহায্যে আপনি একাধিক concurrent requests (একযোগে অনুরোধ) হ্যান্ডল করতে পারেন, যার ফলে পারফরম্যান্স উন্নত হয় এবং সার্ভারের প্রতি লোড কমে। এখানে আলোচনা করা হবে কিভাবে concurrent requests হ্যান্ডল করা যায় এবং FastAPI-তে performance optimization করা যায়।

Concurrent Requests in FastAPI

FastAPI তে concurrent requests হ্যান্ডল করতে asynchronous ফাংশন ব্যবহৃত হয়। সাধারণত, async এবং await কীওয়ার্ড ব্যবহার করে asynchronous programming করা হয়, যা একাধিক রিকোয়েস্ট একই সময়ে প্রসেস করতে সক্ষম হয়।

Asynchronous Programming

FastAPI তে, যদি আপনি কোন ফাংশনকে asynchronous হিসাবে ডিফাইন করেন, তা non-blocking ভাবে কাজ করবে, অর্থাৎ, একটি রিকোয়েস্ট সম্পন্ন না হওয়া পর্যন্ত অন্য রিকোয়েস্ট অপেক্ষা করবে না। এর ফলে, সার্ভার একাধিক রিকোয়েস্টকে параллেল ভাবে প্রসেস করতে সক্ষম হয়।

উদাহরণ: Concurrent Requests (Asynchronous Function)

from fastapi import FastAPI
import asyncio

app = FastAPI()

# একটি asynchronous function
async def fake_data_processing():
    await asyncio.sleep(2)  # 2 সেকেন্ড সময় নিবে
    return "Data Processed"

@app.get("/process/")
async def process_data():
    result = await fake_data_processing()  # Fake Data Processing
    return {"message": result}

এখানে, fake_data_processing() ফাংশনটি 2 সেকেন্ডের জন্য sleep করবে, তবে এটি non-blocking হিসেবে কাজ করবে, এবং অন্য রিকোয়েস্ট একই সময়ে প্রসেস করা সম্ভব হবে।

Concurrent Requests Example

এখন যদি আপনি এই এন্ডপয়েন্টটি একাধিকবার কল করেন, FastAPI একযোগে সমস্ত রিকোয়েস্ট হ্যান্ডল করবে এবং সেগুলি একে অপরের জন্য ব্লক হবে না।

1. GET /process/ রিকোয়েস্টের জন্য প্রথম কল।
2. একই সময়ে GET /process/ রিকোয়েস্টের জন্য দ্বিতীয় কল।

এভাবে asynchronous ফাংশন ব্যবহার করে একাধিক রিকোয়েস্ট হ্যান্ডল করতে পারবেন এবং এটি সার্ভারের পারফরম্যান্স বাড়াবে।

Performance Optimization in FastAPI

FastAPI ডিফল্টভাবে অনেক উন্নত পারফরম্যান্স প্রদান করে, তবে কিছু কৌশল রয়েছে যেগুলির মাধ্যমে আরও পারফরম্যান্স উন্নত করা যেতে পারে।

1. Asynchronous I/O Operations

FastAPI তে asynchronous I/O অপারেশন ব্যবহার করলে আপনার অ্যাপ্লিকেশন দ্রুত এবং বেশি রিকোয়েস্ট হ্যান্ডল করতে সক্ষম হবে। I/O অপারেশন যেমন ডাটাবেস অ্যাক্সেস, ফাইল সিস্টেম অপারেশন এবং API কল সাধারণত সিঙ্ক্রোনাস হলেও, আপনি এগুলি asynchronousভাবে করার মাধ্যমে পারফরম্যান্স বৃদ্ধি করতে পারেন।

উদাহরণ: Asynchronous Database Query

from fastapi import FastAPI
import asyncio

app = FastAPI()

# asynchronous database query simulation
async def get_data_from_db():
    await asyncio.sleep(1)  # simulating async database query
    return {"data": "Fake Data from DB"}

@app.get("/fetch_data/")
async def fetch_data():
    data = await get_data_from_db()  # Asynchronous DB Query
    return data

এখানে asyncio.sleep(1) দিয়ে আমরা ফেক ডাটাবেস কুয়েরি সিমুলেট করছি, তবে মূল উদ্দেশ্য হল যে, এটি asynchronous ভাবে কাজ করছে, এবং অন্যান্য রিকোয়েস্ট একই সময়ে প্রসেস করা হচ্ছে।

2. Connection Pooling

ডাটাবেস সংযোগের ক্ষেত্রে, connection pooling ব্যবহার করলে ডাটাবেসের সাথে সংযোগ স্থাপন ও বিচ্ছিন্ন করার সময় কমিয়ে আনা যায়। Python-এ asyncpg, databases বা SQLAlchemy এর মতো লাইব্রেরি ব্যবহার করে আপনি কনকশন পুলিং কনফিগার করতে পারেন।

উদাহরণ: SQLAlchemy এবং Connection Pooling

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"  # SQLite connection with async support

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}, pool_size=20)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

@app.get("/items/")
async def get_items():
    db = SessionLocal()
    # Querying the database asynchronously
    db.close()
    return {"message": "Fetched Items"}

এখানে SQLAlchemy লাইব্রেরি ব্যবহার করে কনকশন পুলিং করা হয়েছে, যার ফলে ডাটাবেস কুয়েরি করার সময় পারফরম্যান্স বাড়ানো সম্ভব।

3. Caching

FastAPI অ্যাপ্লিকেশনের পারফরম্যান্স বাড়ানোর জন্য caching ব্যবহার করা খুবই কার্যকরী। ডাটা পুনরায় ব্যবহার করার জন্য Redis বা Memcached এর মতো caching সিস্টেম ব্যবহার করতে পারেন।

উদাহরণ: Redis Caching

import redis
from fastapi import FastAPI

app = FastAPI()

# Redis connection setup
r = redis.Redis(host='localhost', port=6379, db=0)

@app.get("/cache/")
async def get_cache():
    cache_data = r.get('cached_data')  # Check cache for data
    if cache_data:
        return {"message": "Data from Cache", "data": cache_data.decode('utf-8')}
    else:
        # Simulating a DB fetch and caching it
        r.set('cached_data', 'Fetched from DB')
        return {"message": "Data from DB"}

Redis ব্যবহার করে আমরা ডাটার জন্য কাঁচে তৈরি করেছি, যাতে ডাটাবেস কুয়েরি পুনরায় না করতে হয় এবং পারফরম্যান্স উন্নত হয়।

4. Uvicorn Configuration for Production

প্রোডাকশন পরিবেশে FastAPI অ্যাপ্লিকেশন চালানোর সময় Uvicorn-এর কনফিগারেশন খুবই গুরুত্বপূর্ণ। workers সংখ্যা এবং --reload অপশনটি পরিচালনা করা উচিৎ যাতে অ্যাপ্লিকেশন দ্রুত কাজ করে।

উদাহরণ: Uvicorn কনফিগারেশন

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

এখানে --workers ফ্ল্যাগের মাধ্যমে Uvicorn সার্ভারে একাধিক প্রসেস ব্যবহার করা হয়েছে, যা multithreading সুবিধা প্রদান করে এবং অধিক সংখ্যক রিকোয়েস্টকে একযোগে প্রসেস করতে সক্ষম হয়।

5. Compression (GZIP)

GZIP বা অন্যান্য কম্প্রেশন সিস্টেম ব্যবহার করে আপনি রেসপন্স সাইজ কমাতে পারেন, যা রেসপন্স টাইম কমায় এবং ব্যান্ডউইথ সাশ্রয় করে।

উদাহরণ: Compression (GZIP)

from fastapi import FastAPI
from fastapi.responses import JSONResponse
import json

app = FastAPI()

@app.get("/compressed/")
def get_compressed_data():
    data = {"message": "This is a large response payload"}
    return JSONResponse(content=json.dumps(data), headers={"Content-Encoding": "gzip"})

এখানে, Content-Encoding: gzip হেডারটি যোগ করা হয়েছে, যা রেসপন্স কম্প্রেস করবে এবং দ্রুত ডেলিভারি নিশ্চিত করবে।

FastAPI তে concurrent requests হ্যান্ডল করতে asynchronous programming এর মাধ্যমে একযোগে বহু রিকোয়েস্ট প্রসেস করা যায়। এর সাথে performance optimization এর জন্য বিভিন্ন কৌশল যেমন asynchronous I/O, connection pooling, caching, এবং GZIP compression ব্যবহার করা যেতে পারে। এই পদ্ধতিগুলির মাধ্যমে আপনি আপনার FastAPI অ্যাপ্লিকেশনের পারফরম্যান্স বৃদ্ধি করতে পারেন, যাতে এটি অধিক স্কেলেবিলিটি এবং দ্রুত রেসপন্স প্রদান করে।